package com.leyantech.ecp.agenx.common;

import com.leyantech.ecp.agenx.common.di.CuratorFactory;
import com.leyantech.ecp.agenx.common.di.VerticleModule;
import com.leyantech.utility.Logger;
import com.leyantech.utility.NetUtils;
import com.leyantech.utility.statsd.StatsDClientFactory;
import com.leyantech.utility.statsd.StatsDMetricsReporter;

import com.codahale.metrics.MetricRegistry;
import com.google.common.collect.Lists;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Module;
import io.opentracing.util.GlobalTracer;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Closeable;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.VertxOptions;
import io.vertx.core.eventbus.EventBusOptions;
import io.vertx.core.impl.VertxBuilder;
import io.vertx.core.impl.VertxInternal;
import io.vertx.core.spi.cluster.ClusterManager;
import io.vertx.ext.dropwizard.DropwizardMetricsOptions;
import io.vertx.ext.healthchecks.HealthChecks;
import io.vertx.spi.cluster.zookeeper.ZookeeperClusterManager;
import io.vertx.tracing.opentracing.OpenTracingOptions;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.core.LoggerContext;
import org.apache.logging.log4j.core.config.Configurator;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-10-31.
 */
public abstract class ClusterVerticle extends AbstractVerticle {

  protected final Logger log = new Logger(getClass());

  protected static final Logger LOGGER = new Logger(ClusterVerticle.class);

  static {
    // 提前加载 gc 打点上报
    StatsDClientFactory.getClient();
  }

  protected void configHealthCheck(Consumer<HealthChecks> consumer) {
    HealthChecks healthChecks = HealthChecks.create(vertx);
    consumer.accept(healthChecks);
  }

  protected void addRunnableCloseHook(Runnable hook) {
    addCloseHook(promise -> {
      try {
        hook.run();
        promise.complete();
      } catch (Throwable e) {
        promise.fail(e);
      }
    });
  }

  protected void addCloseHook(Closeable hook) {
    ((VertxInternal) (vertx)).addCloseHook(hook);
  }

  protected void removeCloseHook(Closeable hook) {
    ((VertxInternal) (vertx)).addCloseHook(hook);
  }

  protected List<Module> dependentModules() {
    return Collections.emptyList();
  }

  /**
   * liteMember 不存储分布式数据，加入或离开集群节点时会更加的快速，对于启动、停机和重启更友好.
   *
   * @see <a href="https://vertx.io/docs/vertx-hazelcast/java/#_using_lite_members">using_lite_members</a>
   * @see <a href="https://docs.hazelcast.com/imdg/4.2/management/cluster-utilities#enabling-lite-members">enabling-lite-members</a>
   */
  protected boolean isLiteMember() {
    return true;
  }

  /**
   * <pre>
   *   集群管理器，为 null 则不加入集群。注意：加入集群会有额外的内存和线程消耗，所以非必要不加入。
   *    hazelcast 为分布式内存集群管理器，所以内存占用较高。本来首选 zookeeper 作为集群管理器，
   *    但是由于测试和生产环境的 zk 集群版本较低(3.4.8--1),无法和 vertx 4.x 版本兼容，只能选择 hazelcast.
   * </pre>
   */
//  protected ClusterManager configClusterManager() {
//    final String appName = Constructs.LAIN_APPNAME;
//    if (appName == null) {
//      // local
//      return new HazelcastClusterManager();
//    }
//
//    final String serviceDNS = String.join(".", appName, "lain");
//
//    Config config = ConfigUtil.loadConfig();
//
//    config.setLiteMember(isLiteMember());
//
//    config.setClusterName(appName);
//    config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
//    config.setProperty(ClusterProperty.DISCOVERY_SPI_ENABLED.getName(), "true");
//
//    DiscoveryStrategyConfig discoveryStrategyConfig = new DiscoveryStrategyConfig()
//        .setDiscoveryStrategyFactory(new HazelcastKubernetesDiscoveryStrategyFactory())
//        .addProperty(KubernetesProperties.SERVICE_DNS.key(), serviceDNS);
//
//    config.getNetworkConfig()
//        .getJoin()
//        .getDiscoveryConfig()
//        .addDiscoveryStrategyConfig(discoveryStrategyConfig);
//
//    return new HazelcastClusterManager(Hazelcast.newHazelcastInstance(config));
//  }

  /**
   * 返回 null 则不需要加入集群，只需要 connector 和 dispatcher 加入集群就可以.
   */
  protected ClusterManager configClusterManager() {
    return new ZookeeperClusterManager(CuratorFactory.curatorFramework);
  }

  VertxOptions configVertxOptions() {
    final ClusterManager clusterManager = configClusterManager();

    final VertxOptions options = new VertxOptions().setClusterManager(clusterManager);
    options.setTracingOptions(new OpenTracingOptions(GlobalTracer.get()));

    final MetricRegistry metricRegistry = new MetricRegistry();

    options.setMetricsOptions(
        new DropwizardMetricsOptions()
            .setBaseName("")
            .setEnabled(true)
            .setMetricRegistry(metricRegistry)
    );

    StatsDMetricsReporter.forRegistry(metricRegistry)
        .prefixedWith("vertx")
        .build()
        .start(10, TimeUnit.SECONDS);

    final String address = NetUtils.getLocalAddress().getHostAddress();

    options.setEventBusOptions(new EventBusOptions().setHost(address).setPort(5701));

    options.setPreferNativeTransport(true);

    return options;
  }

  protected void setupGuice() {
    final List<Module> modules = Lists.newArrayList(dependentModules());
    modules.add(new VerticleModule(vertx));

    final Injector injector = Guice.createInjector(modules);

    injector.injectMembers(this);
  }

  private void addShutdownHook() {
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      log.info("SHUTDOWN hooked");
      this.vertx.close()
          .onComplete(ignore -> {
            log.info("SHUTDOWN DONE");
            // shutdown log4j2
            if (LogManager.getContext() instanceof LoggerContext) {
              log.info("Shutting down log4j2");
              Configurator.shutdown((LoggerContext) LogManager.getContext());
            } else {
              log.warn("Unable to shutdown log4j2");
            }
          })
          .toCompletionStage()
          .toCompletableFuture()
          .join();
    }));
  }

  public void startVerticle() {

    final VertxOptions vertxOptions = configVertxOptions();
    final VertxBuilder vertxBuilder = new VertxBuilder(vertxOptions).init();

    final Runnable start = () -> {
      final String simpleName = getClass().getSimpleName();
      vertx.executeBlocking(promise -> {
        setupGuice();
        promise.complete();
      })
          .onSuccess(Void ->
              vertx.deployVerticle(this, new DeploymentOptions())
                  .onSuccess(deploymentID -> {
                    addShutdownHook();
                    LOGGER.info("deployVerticle {} id {} succeed.", simpleName, deploymentID);
                  })
                  .onFailure(tx -> LOGGER.error("deployVerticle {} failed.", simpleName, tx))
          )
          .onFailure(throwable -> LOGGER.fatal("start verticle failed.", throwable));
    };

    if (vertxOptions.getClusterManager() == null) {
      this.vertx = vertxBuilder.vertx();
      start.run();
    } else {
      vertxBuilder.clusteredVertx(res -> {
        if (res.succeeded()) {
          this.vertx = res.result();
          start.run();
        } else {
          LOGGER.fatal("start verticle failed.", res.cause());
        }
      });
    }

  }

  public static void main(String[] args) {
    try {
      ((ClusterVerticle) Class.forName(args[0]).newInstance()).startVerticle();
    } catch (Throwable t) {
      LOGGER.error("start verticle failed.", t);
    }
  }

}
